package Concurrency2.Chapter08_Pattern.master_worker;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Instruction：
 * Author：@author MaLi
 */
public class Worker<T extends Task,R> {
    //接收任务的阻塞队列
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    private static AtomicInteger index = new AtomicInteger(1);

    // worker编号
    private int workerId;

    // 执行任务的线程
    private Thread thread = null;

    // 创建worker即启动一个线程执行任务队列中的任务 : 这里直接执行run是一样的
    public Worker() {
        this.workerId = index.getAndIncrement();
        thread = new Thread(()->this.run());
        thread.start();
    }

    //轮询执行任务- 创建worker即执行run方法
    public void run() {
        for (; ; ) {
            try {
                T task = taskQueue.take();
                task.setWorkerId(workerId);
                task.execute(); // 不断在阻塞队列中取出Task执行
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    //提交任务到异步队列
    public void submit(T task, Consumer<R> action) {
        task.resultAction = action; //设置task的回调函数
        try {
            this.taskQueue.put(task);//添加任务到阻塞任务队列.
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
